/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.namesrv.routeinfo;

import com.google.common.collect.Sets;
import io.netty.channel.Channel;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.HashSet;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.common.MixAll;
import org.apache.rocketmq.common.TopicConfig;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.constant.PermName;
import org.apache.rocketmq.common.namesrv.NamesrvConfig;
import org.apache.rocketmq.common.sysflag.TopicSysFlag;
import org.apache.rocketmq.common.topic.TopicValidator;
import org.apache.rocketmq.common.utils.ConcurrentHashMapUtils;
import org.apache.rocketmq.logging.org.slf4j.Logger;
import org.apache.rocketmq.logging.org.slf4j.LoggerFactory;
import org.apache.rocketmq.namesrv.NamesrvController;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingConnectException;
import org.apache.rocketmq.remoting.exception.RemotingSendRequestException;
import org.apache.rocketmq.remoting.exception.RemotingTimeoutException;
import org.apache.rocketmq.remoting.exception.RemotingTooMuchRequestException;
import org.apache.rocketmq.remoting.protocol.DataVersion;
import org.apache.rocketmq.remoting.protocol.RemotingCommand;
import org.apache.rocketmq.remoting.protocol.RequestCode;
import org.apache.rocketmq.remoting.protocol.body.BrokerMemberGroup;
import org.apache.rocketmq.remoting.protocol.body.ClusterInfo;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigAndMappingSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicConfigSerializeWrapper;
import org.apache.rocketmq.remoting.protocol.body.TopicList;
import org.apache.rocketmq.remoting.protocol.header.NotifyMinBrokerIdChangeRequestHeader;
import org.apache.rocketmq.remoting.protocol.header.namesrv.UnRegisterBrokerRequestHeader;
import org.apache.rocketmq.remoting.protocol.namesrv.RegisterBrokerResult;
import org.apache.rocketmq.remoting.protocol.route.BrokerData;
import org.apache.rocketmq.remoting.protocol.route.QueueData;
import org.apache.rocketmq.remoting.protocol.route.TopicRouteData;
import org.apache.rocketmq.remoting.protocol.statictopic.TopicQueueMappingInfo;

/**
 * 路由信息管理器
 */
public class RouteInfoManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.NAMESRV_LOGGER_NAME);
    private final static long DEFAULT_BROKER_CHANNEL_EXPIRED_TIME = 1000 * 60 * 2;

    /**
     * 存储集群元数据的集合，用读写锁进行保护，确保数据数据安全。
     */
    private final ReadWriteLock lock = new ReentrantReadWriteLock();

    /**
     * key是主题名称，value是map集合，key是brokerName， value是主题消息队列的元数据（主题所在broker名称，读队列个数，写队列个数，权限位，主题的系统标志）
     * 相同brokername的有很多brokerid不一样的broker，一个brokername对应多个broker，是有主从节点的
     */
    private final Map<String/* topic */, Map<String, QueueData>> topicQueueTable;

    /**
     * key是broker名称，value是broker元数据，包括了broker所属的集群名称（主从复制模式），broker名称，以及broker名称下的主从节点实例地址。
     * name是broker-A，有三个name是一样的节点，id分别是broker-0、broker-1、broker-2
     */
    private final Map<String/* brokerName */, BrokerData> brokerAddrTable;

    /**
     * key是集群名称，value是集群中broker名称集合
     */
    private final Map<String/* clusterName */, Set<String/* brokerName */>> clusterAddrTable;

    /**
     * key是broker地址信息（包括broker所属的集群名称，borker的地址字符串），
     * value是Broker活跃信息（包括上次更新时间戳，心跳超时时间（毫秒），数据版本，netty通道实例（broker和nameserver保持长连接，有一个通道做对应，这样就可以随时通信），高可用服务节点地址）
     */
    private final Map<BrokerAddrInfo/* brokerAddr */, BrokerLiveInfo> brokerLiveTable;

    /**
     * key是broker地址信息（包括broker所属的集群名称，broker的地址字符串），value是过滤器名称列表
     */
    private final Map<BrokerAddrInfo/* brokerAddr */, List<String>/* Filter Server */> filterServerTable;

    /**
     * key是主题名称，
     * value是主题队列映射信息（包含了broker名称（主从关系），主题名称，主题在该broker上的队列总数，用于隔离旧的脏数据的纪元数字等）
     * 纪元数字：broker宕机了后来启动起来了，会有旧数据、broker汇报信息的时候宕机了，但是又活了，会拿着旧数据发送。通过纪元数字来记录，出现一次事故，就往上增长一次，通过纪元数字来隔离旧数据
     */
    private final Map<String/* topic */, Map<String/*brokerName*/, TopicQueueMappingInfo>> topicQueueMappingInfoTable;

    private final BatchUnregistrationService unRegisterService;

    /**
     * 命名服务器控制器
     */
    private final NamesrvController namesrvController;
    private final NamesrvConfig namesrvConfig;

    public RouteInfoManager(final NamesrvConfig namesrvConfig, NamesrvController namesrvController) {
        this.topicQueueTable = new ConcurrentHashMap<>(1024); // 主题队列表
        this.brokerAddrTable = new ConcurrentHashMap<>(128); // broker 地址表
        this.clusterAddrTable = new ConcurrentHashMap<>(32); // 集群地址表
        // 并发map集合，上百个broker向nameserver放信息，每个都有一个长连接，多个死掉了，需要用多线程的方式修改brokerLiveTable数据
        this.brokerLiveTable = new ConcurrentHashMap<>(256); // 活跃 broker 表
        this.filterServerTable = new ConcurrentHashMap<>(256); // 过滤器服务表
        this.topicQueueMappingInfoTable = new ConcurrentHashMap<>(1024); // 主题队列映射表
        this.unRegisterService = new BatchUnregistrationService(this, namesrvConfig);
        this.namesrvConfig = namesrvConfig;
        this.namesrvController = namesrvController;
    }

    public void start() {
        // 启动BatchUnregistrationService服务
        this.unRegisterService.start();
    }

    public void shutdown() {
        // 关闭BatchUnregistrationService服务
        this.unRegisterService.shutdown(true);
    }

    public boolean submitUnRegisterBrokerRequest(UnRegisterBrokerRequestHeader unRegisterRequest) {
        return this.unRegisterService.submit(unRegisterRequest);
    }

    // For test only
    int blockedUnRegisterRequests() {
        return this.unRegisterService.queueLength();
    }

    /**
     * 获取所有集群信息
     * @return
     */
    public ClusterInfo getAllClusterInfo() {
        // 创建ClusterInfo对象
        ClusterInfo clusterInfoSerializeWrapper = new ClusterInfo();
        // 设置broker地址表
        clusterInfoSerializeWrapper.setBrokerAddrTable(this.brokerAddrTable);
        // 设置集群地址表
        clusterInfoSerializeWrapper.setClusterAddrTable(this.clusterAddrTable);
        // 返回集群信息对象
        return clusterInfoSerializeWrapper;
    }

    /**
     * 注册主题信息，操作的是topicQueueTable：
     *  - 该表的key是主题名称；
     *  - value是包含了该主题消息队列所在broker名称以及消息队列元数据的QueueData对象。
     *
     * @param topic 主题名称
     * @param queueDatas 主题消息队列元数据集合，包含了主题队列元数据和队列所在broker的名称。
     */
    public void registerTopic(final String topic, List<QueueData> queueDatas) {
        // 如果队列数据为空，则直接返回
        if (queueDatas == null || queueDatas.isEmpty()) {
            return;
        }

        try {
            // 获取写锁
            this.lock.writeLock().lockInterruptibly();
            // 如果主题队列表中包含了当前主题的信息，则
            if (this.topicQueueTable.containsKey(topic)) {
                // 获取当前主题的主题队列集合
                Map<String, QueueData> queueDataMap  = this.topicQueueTable.get(topic);
                // 遍历主题的主题队列集合
                for (QueueData queueData : queueDatas) {
                    // 如果broker地址表中不包含当前队列所在的broker，表示broker离线，记录警告信息，主题包含错误数据，方法返回。
                    if (!this.brokerAddrTable.containsKey(queueData.getBrokerName())) {
                        log.warn("Register topic contains illegal broker, {}, {}", topic, queueData);
                        return;
                    }
                    queueDataMap.put(queueData.getBrokerName(), queueData);
                }
                log.info("Topic route already exist.{}, {}", topic, this.topicQueueTable.get(topic));
            } else {
                // 如果主题队列表中不包含当前主题信息，则创建新的HashMap，用于保存当前主题的消息队列信息
                Map<String, QueueData> queueDataMap = new HashMap<>();
                // 遍历当前主题的队列元数据信息
                for (QueueData queueData : queueDatas) {
                    // 如果broker地址表中不包含当前队列所在的broker，表示broker离线，记录警告信息，主题包含错误数据，方法返回。
                    if (!this.brokerAddrTable.containsKey(queueData.getBrokerName())) {
                        log.warn("Register topic contains illegal broker, {}, {}", topic, queueData);
                        return;
                    }
                    // 将broker名称作为key，该broker上当前主题的队列信息保存在队列数据表中
                    queueDataMap.put(queueData.getBrokerName(), queueData);
                }
                // 以主题名称为key，队列数据表作为value，添加到主题队列表中。
                this.topicQueueTable.put(topic, queueDataMap);
                log.info("Register topic route:{}, {}", topic, queueDatas);
            }
        } catch (Exception e) {
            log.error("registerTopic Exception", e);
        } finally {
            // 释放写锁
            this.lock.writeLock().unlock();
        }
    }

    public void deleteTopic(final String topic) {
        try {
            this.lock.writeLock().lockInterruptibly();
            this.topicQueueTable.remove(topic);
        } catch (Exception e) {
            log.error("deleteTopic Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public void deleteTopic(final String topic, final String clusterName) {
        try {
            this.lock.writeLock().lockInterruptibly();
            //get all the brokerNames fot the specified cluster
            Set<String> brokerNames = this.clusterAddrTable.get(clusterName);
            if (brokerNames == null || brokerNames.isEmpty()) {
                return;
            }
            //get the store information for single topic
            Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
            if (queueDataMap != null) {
                for (String brokerName : brokerNames) {
                    final QueueData removedQD = queueDataMap.remove(brokerName);
                    if (removedQD != null) {
                        log.info("deleteTopic, remove one broker's topic {} {} {}", brokerName, topic, removedQD);
                    }
                }
                if (queueDataMap.isEmpty()) {
                    log.info("deleteTopic, remove the topic all queue {} {}", clusterName, topic);
                    this.topicQueueTable.remove(topic);
                }
            }
        } catch (Exception e) {
            log.error("deleteTopic Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    public TopicList getAllTopicList() {
        TopicList topicList = new TopicList();
        try {
            this.lock.readLock().lockInterruptibly();
            topicList.getTopicList().addAll(this.topicQueueTable.keySet());
        } catch (Exception e) {
            log.error("getAllTopicList Exception", e);
        } finally {
            this.lock.readLock().unlock();
        }

        return topicList;
    }

    public RegisterBrokerResult registerBroker(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId,
        final String haServerAddr,
        final String zoneName,
        final Long timeoutMillis,
        final TopicConfigSerializeWrapper topicConfigWrapper,
        final List<String> filterServerList,
        final Channel channel) {
        return registerBroker(clusterName, brokerAddr, brokerName, brokerId, haServerAddr, zoneName, timeoutMillis, false, topicConfigWrapper, filterServerList, channel);
    }

    /**
     * 执行broker注册
     * @param clusterName         集群名称
     * @param brokerAddr          broker地址
     * @param brokerName          broker名称
     * @param brokerId            brokerId
     * @param haServerAddr        broekr的HA服务器地址
     * @param zoneName            broker所在区域名称
     * @param timeoutMillis       处理超时时间
     * @param enableActingMaster  是否启用扮演master
     * @param topicConfigWrapper  主题配置信息封装对象
     * @param filterServerList    过滤器服务器列表
     * @param channel             NameServer与broker连接的Netty通道
     * @return 返回注册的结果
     */
    public RegisterBrokerResult registerBroker(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId,
        final String haServerAddr,
        final String zoneName,
        final Long timeoutMillis,
        final Boolean enableActingMaster,
        final TopicConfigSerializeWrapper topicConfigWrapper,
        final List<String> filterServerList,
        final Channel channel) {
        RegisterBrokerResult result = new RegisterBrokerResult();
        try {
            // 获取写锁，会有很多个 broker 发请求，过来注册，所以要保证线程安全
            this.lock.writeLock().lockInterruptibly();

            // 从clusterAddrTable表中获取指定clusterName对应的broker名称集合；
            // 如果没有，则创建新的HashSet集合，以clusterName作为key，新的Set集合作为value存储到clusterAddrTable中。
            // 如果 broker 第一次注册，那么就从 clusterAddrTable 里面找不到指定的 clusterName
            Set<String> brokerNames = ConcurrentHashMapUtils.computeIfAbsent((ConcurrentHashMap<String, Set<String>>) this.clusterAddrTable, clusterName, k -> new HashSet<>());
            // 将参数的broker名称添加到当前集群的broker名称列表中
            brokerNames.add(brokerName);

            boolean registerFirst = false;

            // 从broker地址表中获取指定broker名称的broker数据信息
            // （该类表示典型的broker集群的详细信息（主从复制）：所属的集群名称，broker名称，以及集群中borker名称下所有实例的集合。 同时也包括了zoneName信息。）
            BrokerData brokerData = this.brokerAddrTable.get(brokerName);
            // 如果是null值，表示该broker名称的broker第一次注册
            if (null == brokerData) {
                registerFirst = true;
                // 新建brokerData对象
                brokerData = new BrokerData(clusterName, brokerName, new HashMap<>());
                // 将broker数据对象添加到broker地址表中
                this.brokerAddrTable.put(brokerName, brokerData);
            }

            boolean isOldVersionBroker = enableActingMaster == null;
            // 赋值
            brokerData.setEnableActingMaster(!isOldVersionBroker && enableActingMaster);
            // 赋值
            brokerData.setZoneName(zoneName);

            // 获取broker的地址集合，key是brokerId，value是broker的地址
            Map<Long, String> brokerAddrsMap = brokerData.getBrokerAddrs();

            boolean isMinBrokerIdChanged = false;
            long prevMinBrokerId = 0;
            // 如果broker地址表不是空，则获取其最小的brokerId
            if (!brokerAddrsMap.isEmpty()) {
                prevMinBrokerId = Collections.min(brokerAddrsMap.keySet());
            }
            // 如果当前注册的brokerId比之前的最小ID还小，表示最小brokerId 发生变化，比如当前启动的是master broker
            // 比如原来有的 brokerID 是 2 3 4，但是都不是 master，如果现在注册的是 0 的，也就是 master，那么就把 isMinBrokerIdChanged 设置成 true
            if (brokerId < prevMinBrokerId) {
                isMinBrokerIdChanged = true;
            }

            // 在broker地址表中，如果发生slave到master的角色转换，则需要首先移除该broker节点地址和brokerId的条目，然后再添加该broker节点地址和masterId的条目。就是同样的地址，原先是 slave，现在注册的是 master
            // 即如果broker地址相同，但是brokerId不同，则需要先删除旧的数据，再添加新的数据
            brokerAddrsMap.entrySet().removeIf(item -> null != brokerAddr && brokerAddr.equals(item.getValue()) && brokerId != item.getKey());

            // 对比当前broker的状态版本，如果当前broker的状态版本小于nameserver中保存的数据的状态版本，说明当前元数据旧，拒绝处理即可。
            String oldBrokerAddr = brokerAddrsMap.get(brokerId);
            if (null != oldBrokerAddr && !oldBrokerAddr.equals(brokerAddr)) { // 地址不一样了
                // 根据broker的地址信息获取broker的活跃信息
                BrokerLiveInfo oldBrokerInfo = brokerLiveTable.get(new BrokerAddrInfo(clusterName, oldBrokerAddr));

                if (null != oldBrokerInfo) {
                    // 获取状态版本
                    long oldStateVersion = oldBrokerInfo.getDataVersion().getStateVersion();
                    // 从参数中获取新的状态版本
                    long newStateVersion = topicConfigWrapper.getDataVersion().getStateVersion();
                    // 如果旧数据的状态版本大于新数据的状态版本，则拒绝处理，同时将当前broker的信息从brokerLiveTable中移除
                    if (oldStateVersion > newStateVersion) {
                        log.warn("Registered Broker conflicts with the existed one, just ignore.: Cluster:{}, BrokerName:{}, BrokerId:{}, " +
                                "Old BrokerAddr:{}, Old Version:{}, New BrokerAddr:{}, New Version:{}.",
                            clusterName, brokerName, brokerId, oldBrokerAddr, oldStateVersion, brokerAddr, newStateVersion);
                        // 同时将当前broker的信息从brokerLiveTable中移除
                        brokerLiveTable.remove(new BrokerAddrInfo(clusterName, brokerAddr));
                        return result;
                    }
                }
            }

            // 如果broker地址表中不包含该brokerId，并且有主题配置信息需要注册，则报错：broker都没有完成注册，更别提注册broker上主题的信息了。
            if (!brokerAddrsMap.containsKey(brokerId) && topicConfigWrapper.getTopicConfigTable().size() == 1) {
                log.warn("Can't register topicConfigWrapper={} because broker[{}]={} has not registered.",
                    topicConfigWrapper.getTopicConfigTable(), brokerId, brokerAddr);
                return null;
            }

            // 将当前 broker 信息添加到 broker 地址表中，返回旧的地址信息
            String oldAddr = brokerAddrsMap.put(brokerId, brokerAddr);
            // 是否是第一次注册
            registerFirst = registerFirst || (StringUtils.isEmpty(oldAddr));

            // 如果brokerId是0，则表示注册的broker是master
            boolean isMaster = MixAll.MASTER_ID == brokerId;
            // 是否是首要broker slave：如果不是旧版本的broker，并且不是master，并且brokerId是当前brokerName中id最小的broker
            boolean isPrimeSlave = !isOldVersionBroker && !isMaster
                && brokerId == Collections.min(brokerAddrsMap.keySet());

            // 如果新注册的broker上有主题配置信息需要注册，并且新注册的broker要么是master broker，要么是首要slave，则执行主题信息注册
            if (null != topicConfigWrapper && (isMaster || isPrimeSlave)) {

                // 从参数中获取主题配置表
                ConcurrentMap<String, TopicConfig> tcTable = topicConfigWrapper.getTopicConfigTable();
                // 如果存在主题配置表
                if (tcTable != null) {
                    // 反序列化主题配置数据
                    TopicConfigAndMappingSerializeWrapper mappingSerializeWrapper = TopicConfigAndMappingSerializeWrapper.from(topicConfigWrapper);
                    // 获取主题队列映射表
                    Map<String, TopicQueueMappingInfo> topicQueueMappingInfoMap = mappingSerializeWrapper.getTopicQueueMappingInfoMap();

                    /*
                     * deleteTopicWithBrokerRegistration：默认是 false
                     *
                     * 如果该值设置为true，则如果broker注册的报文体中不包含主题信息，则会导致主题信息从nameserver删除。
                     * 默认为false。
                     * 警告：
                     * 1. 如果要设置为true，则需要同时设置enableSingleTopicRegister为true，防止意外丢失主题路由信息；
                     * 2. 当前该标志不支持静态主题。
                     *
                     * namesrvConfig.isDeleteTopicWithBrokerRegistration() 为 true：在注册的时候，需要删除 broker 信息，那么如果我还没有写，先把原来的删掉了，再写新的，但是新的里面不包含主题信息，那么就会导致主题信息丢失
                     */
                    // 如果启用了deleteTopicWithBrokerRegistration，同时主题队列映射信息表是空的
                    if (namesrvConfig.isDeleteTopicWithBrokerRegistration() && topicQueueMappingInfoMap.isEmpty()) {
                        // 根据参数中的 brokerName 获取该 broker 上关联的主题名称集合
                        final Set<String> oldTopicSet = topicSetOfBrokerName(brokerName);
                        // 从参数中的主题配置表中获取主题名称集合
                        final Set<String> newTopicSet = tcTable.keySet();
                        // 返回旧主题集合中存在的，但在新主题集合中不存在的主题集合
                        final Sets.SetView<String> toDeleteTopics = Sets.difference(oldTopicSet, newTopicSet);
                        // 删除这些不存在于新集合中的主题信息
                        for (final String toDeleteTopic : toDeleteTopics) {
                            // key是broker名称，value是队列信息
                            Map<String, QueueData> queueDataMap = topicQueueTable.get(toDeleteTopic);
                            // 从队列数据中删除对应的消息队列
                            final QueueData removedQD = queueDataMap.remove(brokerName);
                            if (removedQD != null) {
                                log.info("deleteTopic, remove one broker's topic {} {} {}", brokerName, toDeleteTopic, removedQD);
                            }
                            // 如果删除完成之后队列数据表为空，则从主题队列表中删除当前主题条目
                            if (queueDataMap.isEmpty()) {
                                log.info("deleteTopic, remove the topic all queue {}", toDeleteTopic);
                                topicQueueTable.remove(toDeleteTopic);
                            }
                        }
                    }

                    // 遍历主题配置表的条目
                    for (Map.Entry<String, TopicConfig> entry : tcTable.entrySet()) {
                        // 如果是第一次注册该broker
                        // 或者
                        // 主题配置信息更改了
                        if (registerFirst || this.isTopicConfigChanged(clusterName, brokerAddr,
                            topicConfigWrapper.getDataVersion(), brokerName,
                            entry.getValue().getTopicName())) {
                            // 获取主题配置信息
                            final TopicConfig topicConfig = entry.getValue();
                            if (isPrimeSlave) {
                                // 从首要slave broker清除写权限
                                topicConfig.setPerm(topicConfig.getPerm() & (~PermName.PERM_WRITE));
                            }
                            // 创建或更新broker上的主题配置信息
                            this.createAndUpdateQueueData(brokerName, topicConfig);
                        }
                    }

                    // 如果broker上主题配置信息更改了，或者第一次注册broker
                    if (this.isBrokerTopicConfigChanged(clusterName, brokerAddr, topicConfigWrapper.getDataVersion()) || registerFirst) {
                        //the topicQueueMappingInfoMap should never be null, but can be empty
                        // 遍历主题名称到主题队列映射信息的Map集合
                        for (Map.Entry<String, TopicQueueMappingInfo> entry : topicQueueMappingInfoMap.entrySet()) {
                            // 主题队列映射表中是否包含该主题的条目，如果不包含，则
                            if (!topicQueueMappingInfoTable.containsKey(entry.getKey())) {
                                // 向主题队列映射信息表中添加当前主题名称与主题队列映射信息的键值对
                                topicQueueMappingInfoTable.put(entry.getKey(), new HashMap<>());
                            }
                            // 向主题队列映射信息表中添加条目：
                            // 首先根据主题名称，获取map集合，map集合的key是brokerName，value是主题队列映射信息
                            //                           主题名称-> map集合   map集合的key：brokerName    map集合的value：TopicQueueMappingInfo
                            topicQueueMappingInfoTable.get(entry.getKey()).put(entry.getValue().getBname(), entry.getValue());
                        }
                    }
                }
            }

            // 创建新的BrokerAddrInfo对象，封装集群名称，broker地址
            BrokerAddrInfo brokerAddrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
            // 将brokerAddrInfo对象作为key，brokerLiveInfo作为value添加到brokerLiveTable中
            BrokerLiveInfo prevBrokerLiveInfo = this.brokerLiveTable.put(brokerAddrInfo,
                new BrokerLiveInfo(
                    System.currentTimeMillis(),
                    timeoutMillis == null ? DEFAULT_BROKER_CHANNEL_EXPIRED_TIME : timeoutMillis,
                    topicConfigWrapper == null ? new DataVersion() : topicConfigWrapper.getDataVersion(),
                    channel,
                    haServerAddr));
            if (null == prevBrokerLiveInfo) {
                // 如果没有旧值，说明是broker第一次注册
                log.info("new broker registered, {} HAService: {}", brokerAddrInfo, haServerAddr);
            }

            // 如果过滤器服务器列表不是null
            if (filterServerList != null) {
                // 如果内容是空的，则从filterServerTable中移除当前broker的brokerAddrInfo对应的键值对信息
                if (filterServerList.isEmpty()) {
                    this.filterServerTable.remove(brokerAddrInfo);
                } else {
                    // 否则将当前broker的brokerAddrInfo作为key，filterServerList作为value添加到filterServerTable中，有可能是替换。
                    this.filterServerTable.put(brokerAddrInfo, filterServerList);
                }
            }

            // 如果注册的broker不是master
            if (MixAll.MASTER_ID != brokerId) {
                // 获取master的地址
                String masterAddr = brokerData.getBrokerAddrs().get(MixAll.MASTER_ID);
                if (masterAddr != null) {
                    // 如果地址存在，则创建BrokerAddrInfo对象
                    BrokerAddrInfo masterAddrInfo = new BrokerAddrInfo(clusterName, masterAddr);
                    // 从brokerLiveTable中获取brokerLiveInfo信息
                    BrokerLiveInfo masterLiveInfo = this.brokerLiveTable.get(masterAddrInfo);
                    if (masterLiveInfo != null) {
                        // 在结果中设置HaServerAddr为master中记录的HaServerAddr，masterAddr设置为masterAddr即可。
                        result.setHaServerAddr(masterLiveInfo.getHaServerAddr());
                        result.setMasterAddr(masterAddr);
                    }
                }
            }

            // 如果相同brokerName的broker主从集群中，最小ID更改了，并且需要通知最小ID更改事件
            if (isMinBrokerIdChanged && namesrvConfig.isNotifyMinBrokerIdChanged()) {
                // 则通知处理相同brokerName的broker主从集群中，最小ID更改事件
                notifyMinBrokerIdChanged(brokerAddrsMap, null,
                    this.brokerLiveTable.get(brokerAddrInfo).getHaServerAddr());
            }
        } catch (Exception e) {
            log.error("registerBroker Exception", e);
        } finally {
            // 释放写锁
            this.lock.writeLock().unlock();
        }
        // 返回结果
        return result;
    }

    /**
     * 获取与当前broker名称相关的主题名称集合
     * @param brokerName broker名称
     * @return
     */
    private Set<String> topicSetOfBrokerName(final String brokerName) {
        // 创建集合
        Set<String> topicOfBroker = new HashSet<>();
        // 遍历主题队列表
        //    key是主题名称；
        //    value是map集合：
        //      key是brokerName，
        //      value是主题消息队列的元数据（主题所在broker名称，读队列个数，写队列个数，权限位，主题的系统标志）
        for (final Entry<String, Map<String, QueueData>> entry : this.topicQueueTable.entrySet()) {
            // 如果包含brokerName
            if (entry.getValue().containsKey(brokerName)) {
                // 将主题名称添加到结果集合中
                topicOfBroker.add(entry.getKey());
            }
        }
        // 返回结果集合
        return topicOfBroker;
    }

    public BrokerMemberGroup getBrokerMemberGroup(String clusterName, String brokerName) {
        BrokerMemberGroup groupMember = new BrokerMemberGroup(clusterName, brokerName);
        try {
            try {
                this.lock.readLock().lockInterruptibly();
                final BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                if (brokerData != null) {
                    groupMember.getBrokerAddrs().putAll(brokerData.getBrokerAddrs());
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("Get broker member group exception", e);
        }
        return groupMember;
    }

    /**
     * broker主题配置信息是否更新了
     * @param clusterName  集群名称
     * @param brokerAddr   broker地址
     * @param dataVersion  数据版本
     * @return true表示broker数据更新了，false表示两端数据一致。
     */
    public boolean isBrokerTopicConfigChanged(final String clusterName, final String brokerAddr,
        final DataVersion dataVersion) {
        // 查询指定集群，指定broker地址的数据版本
        DataVersion prev = queryBrokerTopicConfig(clusterName, brokerAddr);
        // 如果NameServer端没有该数据版本信息，或者两者不一致，则返回true
        return null == prev || !prev.equals(dataVersion);
    }

    public boolean isTopicConfigChanged(final String clusterName, final String brokerAddr,
        final DataVersion dataVersion, String brokerName, String topic) {
        boolean isChange = isBrokerTopicConfigChanged(clusterName, brokerAddr, dataVersion);
        if (isChange) {
            return true;
        }
        final Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
        if (queueDataMap == null || queueDataMap.isEmpty()) {
            return true;
        }

        // The topicQueueTable already contains the broker
        return !queueDataMap.containsKey(brokerName);
    }

    /**
     * 查询指定集群名称，指定broker地址的数据版本信息并返回
     * @param clusterName  集群名称
     * @param brokerAddr   broker地址
     * @return 返回数据版本对象
     */
    public DataVersion queryBrokerTopicConfig(final String clusterName, final String brokerAddr) {
        BrokerAddrInfo addrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
        /*
         * private final Map<BrokerAddrInfo, BrokerLiveInfo> brokerLiveTable
         * key是broker地址信息（包括broker所属的集群名称，borker的地址字符串），
         * value是Broker活跃信息（包括上次更新时间戳，心跳超时时间（毫秒），数据版本，netty通道实例，高可用服务节点地址）
         */
        BrokerLiveInfo prev = this.brokerLiveTable.get(addrInfo);
        if (prev != null) {
            return prev.getDataVersion();
        }
        return null;
    }

    /**
     * 更新broker信息更新时间戳为当前时间毫秒值
     * @param clusterName  集群名称
     * @param brokerAddr   broker地址
     */
    public void updateBrokerInfoUpdateTimestamp(final String clusterName, final String brokerAddr) {
        // 创建broker地址信息
        BrokerAddrInfo addrInfo = new BrokerAddrInfo(clusterName, brokerAddr);
        // 从broker活跃表中获取对应的broker活跃信息对象
        BrokerLiveInfo prev = this.brokerLiveTable.get(addrInfo);
        if (prev != null) {
            // 设置最近更新时间为当前时间
            prev.setLastUpdateTimestamp(System.currentTimeMillis());
        }
    }

    /**
     *
     * @param brokerName
     * @param topicConfig
     */
    private void createAndUpdateQueueData(final String brokerName, final TopicConfig topicConfig) {
        // 创建新的QueueData对象
        QueueData queueData = new QueueData();
        // 赋值brokerName
        queueData.setBrokerName(brokerName);
        // 赋值写队列个数
        queueData.setWriteQueueNums(topicConfig.getWriteQueueNums());
        // 赋值读队列个数
        queueData.setReadQueueNums(topicConfig.getReadQueueNums());
        // 赋值权限
        queueData.setPerm(topicConfig.getPerm());
        // 设置主题的系统标记
        queueData.setTopicSysFlag(topicConfig.getTopicSysFlag());

        // 从主题队列表中获取指定主题名称对应的map集合，key是brokerName，value是QueueData对象
        Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topicConfig.getTopicName());
        // 如果没有当前主题对应的queueDataMap
        if (null == queueDataMap) {
            // 新建map集合
            queueDataMap = new HashMap<>();
            // 将brokerName作为key，queueData作为value添加到queueDataMap中
            queueDataMap.put(brokerName, queueData);
            // 将主题名称，queueDataMap添加到主题队列表中
            this.topicQueueTable.put(topicConfig.getTopicName(), queueDataMap);
            log.info("new topic registered, {} {}", topicConfig.getTopicName(), queueData);
        } else {
            // 如果有当前主题对应的queueDataMap，则直接获取当前brokerName对应的队列元数据信息
            final QueueData existedQD = queueDataMap.get(brokerName);
            // 如果不存在队列元数据信息
            if (existedQD == null) {
                // 如果没有当前brokerName对应的键值对，则直接将brokerName->queueData添加到queueDataMap中
                queueDataMap.put(brokerName, queueData);
            } else if (!existedQD.equals(queueData)) {
                log.info("topic changed, {} OLD: {} NEW: {}", topicConfig.getTopicName(), existedQD,
                    queueData);
                // 如果有旧的与brokerName对应的队列元数据，则直接替换，与上一个分支唯一的区别就是记录了一条日志
                queueDataMap.put(brokerName, queueData);
            }
        }
    }

    public int wipeWritePermOfBrokerByLock(final String brokerName) {
        try {
            try {
                this.lock.writeLock().lockInterruptibly();
                return operateWritePermOfBroker(brokerName, RequestCode.WIPE_WRITE_PERM_OF_BROKER);
            } finally {
                this.lock.writeLock().unlock();
            }
        } catch (Exception e) {
            log.error("wipeWritePermOfBrokerByLock Exception", e);
        }

        return 0;
    }

    public int addWritePermOfBrokerByLock(final String brokerName) {
        try {
            try {
                this.lock.writeLock().lockInterruptibly();
                return operateWritePermOfBroker(brokerName, RequestCode.ADD_WRITE_PERM_OF_BROKER);
            } finally {
                this.lock.writeLock().unlock();
            }
        } catch (Exception e) {
            log.error("addWritePermOfBrokerByLock Exception", e);
        }
        return 0;
    }

    private int operateWritePermOfBroker(final String brokerName, final int requestCode) {
        int topicCnt = 0;

        for (Entry<String, Map<String, QueueData>> entry : this.topicQueueTable.entrySet()) {
            Map<String, QueueData> qdMap = entry.getValue();

            final QueueData qd = qdMap.get(brokerName);
            if (qd == null) {
                continue;
            }
            int perm = qd.getPerm();
            switch (requestCode) {
                case RequestCode.WIPE_WRITE_PERM_OF_BROKER:
                    perm &= ~PermName.PERM_WRITE;
                    break;
                case RequestCode.ADD_WRITE_PERM_OF_BROKER:
                    perm = PermName.PERM_READ | PermName.PERM_WRITE;
                    break;
            }
            qd.setPerm(perm);
            topicCnt++;
        }
        return topicCnt;
    }

    public void unregisterBroker(
        final String clusterName,
        final String brokerAddr,
        final String brokerName,
        final long brokerId) {
        UnRegisterBrokerRequestHeader unRegisterBrokerRequest = new UnRegisterBrokerRequestHeader();
        unRegisterBrokerRequest.setClusterName(clusterName);
        unRegisterBrokerRequest.setBrokerAddr(brokerAddr);
        unRegisterBrokerRequest.setBrokerName(brokerName);
        unRegisterBrokerRequest.setBrokerId(brokerId);

        unRegisterBroker(Sets.newHashSet(unRegisterBrokerRequest));
    }

    public void unRegisterBroker(Set<UnRegisterBrokerRequestHeader> unRegisterRequests) {
        try {
            Set<String> removedBroker = new HashSet<>();
            Set<String> reducedBroker = new HashSet<>();
            Map<String, BrokerStatusChangeInfo> needNotifyBrokerMap = new HashMap<>();

            this.lock.writeLock().lockInterruptibly();
            for (final UnRegisterBrokerRequestHeader unRegisterRequest : unRegisterRequests) {
                final String brokerName = unRegisterRequest.getBrokerName();
                final String clusterName = unRegisterRequest.getClusterName();
                final String brokerAddr = unRegisterRequest.getBrokerAddr();

                BrokerAddrInfo brokerAddrInfo = new BrokerAddrInfo(clusterName, brokerAddr);

                BrokerLiveInfo brokerLiveInfo = this.brokerLiveTable.remove(brokerAddrInfo);
                log.info("unregisterBroker, remove from brokerLiveTable {}, {}",
                    brokerLiveInfo != null ? "OK" : "Failed",
                    brokerAddrInfo
                );

                this.filterServerTable.remove(brokerAddrInfo);

                boolean removeBrokerName = false;
                boolean isMinBrokerIdChanged = false;
                BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                if (null != brokerData) {
                    if (!brokerData.getBrokerAddrs().isEmpty() &&
                        unRegisterRequest.getBrokerId().equals(Collections.min(brokerData.getBrokerAddrs().keySet()))) {
                        isMinBrokerIdChanged = true;
                    }
                    boolean removed = brokerData.getBrokerAddrs().entrySet().removeIf(item -> item.getValue().equals(brokerAddr));
                    log.info("unregisterBroker, remove addr from brokerAddrTable {}, {}",
                        removed ? "OK" : "Failed",
                        brokerAddrInfo
                    );
                    if (brokerData.getBrokerAddrs().isEmpty()) {
                        this.brokerAddrTable.remove(brokerName);
                        log.info("unregisterBroker, remove name from brokerAddrTable OK, {}",
                            brokerName
                        );

                        removeBrokerName = true;
                    } else if (isMinBrokerIdChanged) {
                        needNotifyBrokerMap.put(brokerName, new BrokerStatusChangeInfo(
                            brokerData.getBrokerAddrs(), brokerAddr, null));
                    }
                }

                if (removeBrokerName) {
                    Set<String> nameSet = this.clusterAddrTable.get(clusterName);
                    if (nameSet != null) {
                        boolean removed = nameSet.remove(brokerName);
                        log.info("unregisterBroker, remove name from clusterAddrTable {}, {}",
                            removed ? "OK" : "Failed",
                            brokerName);

                        if (nameSet.isEmpty()) {
                            this.clusterAddrTable.remove(clusterName);
                            log.info("unregisterBroker, remove cluster from clusterAddrTable {}",
                                clusterName
                            );
                        }
                    }
                    removedBroker.add(brokerName);
                } else {
                    reducedBroker.add(brokerName);
                }
            }

            cleanTopicByUnRegisterRequests(removedBroker, reducedBroker);

            if (!needNotifyBrokerMap.isEmpty() && namesrvConfig.isNotifyMinBrokerIdChanged()) {
                notifyMinBrokerIdChanged(needNotifyBrokerMap);
            }
        } catch (Exception e) {
            log.error("unregisterBroker Exception", e);
        } finally {
            this.lock.writeLock().unlock();
        }
    }

    private void cleanTopicByUnRegisterRequests(Set<String> removedBroker, Set<String> reducedBroker) {
        Iterator<Entry<String, Map<String, QueueData>>> itMap = this.topicQueueTable.entrySet().iterator();
        while (itMap.hasNext()) {
            Entry<String, Map<String, QueueData>> entry = itMap.next();

            String topic = entry.getKey();
            Map<String, QueueData> queueDataMap = entry.getValue();

            for (final String brokerName : removedBroker) {
                final QueueData removedQD = queueDataMap.remove(brokerName);
                if (removedQD != null) {
                    log.debug("removeTopicByBrokerName, remove one broker's topic {} {}", topic, removedQD);
                }
            }

            if (queueDataMap.isEmpty()) {
                log.debug("removeTopicByBrokerName, remove the topic all queue {}", topic);
                itMap.remove();
            }

            for (final String brokerName : reducedBroker) {
                final QueueData queueData = queueDataMap.get(brokerName);

                if (queueData != null) {
                    if (this.brokerAddrTable.get(brokerName).isEnableActingMaster()) {
                        // Master has been unregistered, wipe the write perm
                        if (isNoMasterExists(brokerName)) {
                            queueData.setPerm(queueData.getPerm() & (~PermName.PERM_WRITE));
                        }
                    }
                }
            }
        }
    }

    private boolean isNoMasterExists(String brokerName) {
        final BrokerData brokerData = this.brokerAddrTable.get(brokerName);
        if (brokerData == null) {
            return true;
        }

        if (brokerData.getBrokerAddrs().size() == 0) {
            return true;
        }

        return Collections.min(brokerData.getBrokerAddrs().keySet()) > 0;
    }

    /**
     * 获取指定主题的路由信息。
     * @param topic 需要查询路由信息的主题名称，字符串。
     * @return
     */
    public TopicRouteData pickupTopicRouteData(final String topic) {
        // 创建返回对象
        TopicRouteData topicRouteData = new TopicRouteData();

        // 找到队列元数据的标志
        boolean foundQueueData = false;
        // 找到broker元数据的标志
        boolean foundBrokerData = false;

        // broker元数据列表
        List<BrokerData> brokerDataList = new LinkedList<>();

        // 设置给返回对象
        topicRouteData.setBrokerDatas(brokerDataList);

        // 过滤器服务器集合
        // 主题在broker上的消息队列列表，key是broker名称，value是broker上当前主题的队列元数据列表。
        HashMap<String, List<String>> filterServerMap = new HashMap<>();
        topicRouteData.setFilterServerTable(filterServerMap);

        try {
            // 获取读锁，防止并发修改
            this.lock.readLock().lockInterruptibly();
            // 从主题队列表中获取指定主题的队列集合信息，
            Map<String, QueueData> queueDataMap = this.topicQueueTable.get(topic);
            if (queueDataMap != null) {
                // 设置主题的队列数据
                topicRouteData.setQueueDatas(new ArrayList<>(queueDataMap.values()));
                foundQueueData = true;

                Set<String> brokerNameSet = new HashSet<>(queueDataMap.keySet());
                // 遍历broker
                for (String brokerName : brokerNameSet) {
                    // 从broker地址表中查找broker的master对应的地址
                    BrokerData brokerData = this.brokerAddrTable.get(brokerName);
                    // 找不到跳过
                    if (null == brokerData) {
                        continue;
                    }
                    // 克隆一份数据
                    BrokerData brokerDataClone = new BrokerData(brokerData);
                    // 设置给brokerDataList，该集合已经赋值给主题路由信息对象了
                    brokerDataList.add(brokerDataClone);
                    foundBrokerData = true;
                    // 如果过滤器服务器表为空，则跳过
                    if (filterServerTable.isEmpty()) {
                        continue;
                    }
                    // 否则通过过滤器服务器表查找过滤器服务器列表
                    for (final String brokerAddr : brokerDataClone.getBrokerAddrs().values()) {
                        BrokerAddrInfo brokerAddrInfo = new BrokerAddrInfo(brokerDataClone.getCluster(), brokerAddr);
                        // 找到过滤器服务器列表
                        List<String> filterServerList = this.filterServerTable.get(brokerAddrInfo);
                        // 赋值给过滤器服务器map集合，该集合已经赋值给主题路由信息对象了
                        filterServerMap.put(brokerAddr, filterServerList);
                    }
                }
            }
        } catch (Exception e) {
            log.error("pickupTopicRouteData Exception", e);
        } finally {
            // 释放读锁，允许元数据修改
            this.lock.readLock().unlock();
        }

        log.debug("pickupTopicRouteData {} {}", topic, topicRouteData);

        // 如果指定主题的broker元数据找到，消息队列元数据找到，则
        if (foundBrokerData && foundQueueData) {
            // 从主题队列映射信息表中查找该主题的 主题队列映射信息，并赋值给主题路由数据对象
            topicRouteData.setTopicQueueMappingByBroker(this.topicQueueMappingInfoTable.get(topic));
            // 如果nameserver不支持作为活跃的master，则直接返回主题路由数据对象
            if (!namesrvConfig.isSupportActingMaster()) {
                return topicRouteData;
            }
            // 如果主题名称以"rmq_sys_SYNC_BROKER_MEMBER_"开头，则直接返回主题路由对象数据
            if (topic.startsWith(TopicValidator.SYNC_BROKER_MEMBER_GROUP_PREFIX)) {
                return topicRouteData;
            }
            // 如果broker数据条目为0，或者队列数据条目为0，则直接返回主题路由对象
            if (topicRouteData.getBrokerDatas().size() == 0 || topicRouteData.getQueueDatas().size() == 0) {
                return topicRouteData;
            }

            // 是否作为master
            boolean needActingMaster = false;

            // 遍历broker元数据集合
            for (final BrokerData brokerData : topicRouteData.getBrokerDatas()) {
                // 如果找到了broker，并且没有MASTER节点（主从节点，0表示主节点，其他大于0的值表示从节点）
                if (brokerData.getBrokerAddrs().size() != 0
                    && !brokerData.getBrokerAddrs().containsKey(MixAll.MASTER_ID)) {
                    needActingMaster = true;
                    break;
                }
            }

            // 如果不需要扮演master，则直接返回主题路由数据对象
            if (!needActingMaster) {
                return topicRouteData;
            }

            // 否则遍历broker数据，每个条目对应一个brokerName
            for (final BrokerData brokerData : topicRouteData.getBrokerDatas()) {
                // 拿到broker地址列表
                final HashMap<Long, String> brokerAddrs = brokerData.getBrokerAddrs();
                // 如果没有broker地址，或者它本身就是master，或者broker没有启用扮演master，则跳过
                if (brokerAddrs.size() == 0 || brokerAddrs.containsKey(MixAll.MASTER_ID) || !brokerData.isEnableActingMaster()) {
                    continue;
                }

                // 如果broker没有master
                for (final QueueData queueData : topicRouteData.getQueueDatas()) {
                    // 如果broker名称对应当前brokerName
                    if (queueData.getBrokerName().equals(brokerData.getBrokerName())) {
                        // 且没有写权限
                        if (!PermName.isWriteable(queueData.getPerm())) {
                            // 找最小brokerId的 slave broker
                            final Long minBrokerId = Collections.min(brokerAddrs.keySet());
                            // 从broker地址集合中，移除最小id的slave broker条目
                            final String actingMasterAddr = brokerAddrs.remove(minBrokerId);
                            // 将该broker以master加进入
                            brokerAddrs.put(MixAll.MASTER_ID, actingMasterAddr);
                        }
                        break;
                    }
                }

            }
            // 返回主题路由信息
            return topicRouteData;
        }
        // 如果没有找到指定主题的broker元数据，或者没有找到指定主题队列元数据信息，则返回null值。
        return null;
    }

    public void scanNotActiveBroker() {
        try {
            log.info("start scanNotActiveBroker");
            // brokerLiveTable：活跃 broker 表
            for (Entry<BrokerAddrInfo, BrokerLiveInfo> next : this.brokerLiveTable.entrySet()) {
                long last = next.getValue().getLastUpdateTimestamp(); // 拿到上次更新的时间
                long timeoutMillis = next.getValue().getHeartbeatTimeoutMillis(); // 拿到超时时间
                if ((last + timeoutMillis) < System.currentTimeMillis()) { // 上次更新的时间 + 超时时间小于当前时间，那么就是超时
                    RemotingHelper.closeChannel(next.getValue().getChannel()); // 关闭通道
                    log.warn("The broker channel expired, {} {}ms", next.getKey(), timeoutMillis);
                    this.onChannelDestroy(next.getKey());
                }
            }
        } catch (Exception e) {
            log.error("scanNotActiveBroker exception", e);
        }
    }

    public void onChannelDestroy(BrokerAddrInfo brokerAddrInfo) {
        UnRegisterBrokerRequestHeader unRegisterRequest = new UnRegisterBrokerRequestHeader();
        boolean needUnRegister = false;
        if (brokerAddrInfo != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    needUnRegister = setupUnRegisterRequest(unRegisterRequest, brokerAddrInfo);
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }

        if (needUnRegister) {
            boolean result = this.submitUnRegisterBrokerRequest(unRegisterRequest);
            log.info("the broker's channel destroyed, submit the unregister request at once, " +
                "broker info: {}, submit result: {}", unRegisterRequest, result);
        }
    }

    /**
     * 【broker到nameserver的是长连接】
     * 一旦连接断开，则尝试从BrokerAddrInfo中移除对应的Broker信息。
     *
     *  这是一个 `Channel` 对象的 `onChannelDestroy` 方法，当 `Channel` 对象被销毁时会被调用。
     *  在这个方法中，首先会获取销毁的 `Channel` 对象，然后尝试获取该 `Channel` 对应的 `BrokerAddrInfo` 对象。
     *
     * 如果找到了对应的 `BrokerAddrInfo` 对象，则会调用 `setupUnRegisterRequest` 方法设置 `UnRegisterBrokerRequestHeader` 对象，
     * 然后调用 `submitUnRegisterBrokerRequest` 方法提交该请求。
     *
     * 如果未找到对应的 `BrokerAddrInfo` 对象，则不会进行任何操作。
     *
     * 总之，这个方法是在 `Channel` 对象被销毁时，尝试获取对应的 `BrokerAddrInfo` 对象，
     * 并提交一个取消注册 `Broker` 的请求。
     * @param channel
     */
    public void onChannelDestroy(Channel channel) {
        UnRegisterBrokerRequestHeader unRegisterRequest = new UnRegisterBrokerRequestHeader();
        BrokerAddrInfo brokerAddrFound = null;
        boolean needUnRegister = false;
        if (channel != null) {
            try {
                try {
                    this.lock.readLock().lockInterruptibly();
                    for (Entry<BrokerAddrInfo, BrokerLiveInfo> entry : this.brokerLiveTable.entrySet()) {
                        if (entry.getValue().getChannel() == channel) {
                            brokerAddrFound = entry.getKey();
                            break;
                        }
                    }

                    if (brokerAddrFound != null) {
                        needUnRegister = setupUnRegisterRequest(unRegisterRequest, brokerAddrFound);
                    }
                } finally {
                    this.lock.readLock().unlock();
                }
            } catch (Exception e) {
                log.error("onChannelDestroy Exception", e);
            }
        }

        if (needUnRegister) {
            boolean result = this.submitUnRegisterBrokerRequest(unRegisterRequest);
            log.info("the broker's channel destroyed, submit the unregister request at once, " +
                "broker info: {}, submit result: {}", unRegisterRequest, result);
        }
    }

    private boolean setupUnRegisterRequest(UnRegisterBrokerRequestHeader unRegisterRequest,
        BrokerAddrInfo brokerAddrInfo) {
        unRegisterRequest.setClusterName(brokerAddrInfo.getClusterName());
        unRegisterRequest.setBrokerAddr(brokerAddrInfo.getBrokerAddr());

        for (Entry<String, BrokerData> stringBrokerDataEntry : this.brokerAddrTable.entrySet()) {
            BrokerData brokerData = stringBrokerDataEntry.getValue();
            if (!brokerAddrInfo.getClusterName().equals(brokerData.getCluster())) {
                continue;
            }

            for (Entry<Long, String> entry : brokerData.getBrokerAddrs().entrySet()) {
                Long brokerId = entry.getKey();
                String brokerAddr = entry.getValue();
                if (brokerAddr.equals(brokerAddrInfo.getBrokerAddr())) {
                    unRegisterRequest.setBrokerName(brokerData.getBrokerName());
                    unRegisterRequest.setBrokerId(brokerId);
                    return true;
                }
            }
        }

        return false;
    }

    private void notifyMinBrokerIdChanged(Map<String, BrokerStatusChangeInfo> needNotifyBrokerMap)
        throws InterruptedException, RemotingConnectException, RemotingTimeoutException, RemotingSendRequestException,
        RemotingTooMuchRequestException {
        for (String brokerName : needNotifyBrokerMap.keySet()) {
            BrokerStatusChangeInfo brokerStatusChangeInfo = needNotifyBrokerMap.get(brokerName);
            BrokerData brokerData = brokerAddrTable.get(brokerName);
            if (brokerData != null && brokerData.isEnableActingMaster()) {
                notifyMinBrokerIdChanged(brokerStatusChangeInfo.getBrokerAddrs(),
                    brokerStatusChangeInfo.getOfflineBrokerAddr(), brokerStatusChangeInfo.getHaBrokerAddr());
            }
        }
    }

    /**
     * 通知最小brokerId变更了
     * @param brokerAddrMap        broker地址表
     * @param offlineBrokerAddr    离线broker地址
     * @param haBrokerAddr         habroker地址
     * @throws InterruptedException
     * @throws RemotingSendRequestException
     * @throws RemotingTimeoutException
     * @throws RemotingTooMuchRequestException
     * @throws RemotingConnectException
     */
    private void notifyMinBrokerIdChanged(Map<Long, String> brokerAddrMap, String offlineBrokerAddr,
        String haBrokerAddr)
        throws InterruptedException, RemotingSendRequestException, RemotingTimeoutException,
        RemotingTooMuchRequestException, RemotingConnectException {

        if (brokerAddrMap == null || brokerAddrMap.isEmpty() || this.namesrvController == null) {
            return;
        }

        // 创建通知最小brokerId变更请求报文头
        NotifyMinBrokerIdChangeRequestHeader requestHeader = new NotifyMinBrokerIdChangeRequestHeader();
        // 获取最小的brokerId值
        long minBrokerId = Collections.min(brokerAddrMap.keySet());
        // 设置最小brokerId
        requestHeader.setMinBrokerId(minBrokerId);
        // 设置最小borkerId的broker地址
        requestHeader.setMinBrokerAddr(brokerAddrMap.get(minBrokerId));
        // 设置离线的broker地址
        requestHeader.setOfflineBrokerAddr(offlineBrokerAddr);
        // 设置habroker地址
        requestHeader.setHaBrokerAddr(haBrokerAddr);

        // 选择要通知的broker地址
        List<String> brokerAddrsNotify = chooseBrokerAddrsToNotify(brokerAddrMap, offlineBrokerAddr);
        log.info("min broker id changed to {}, notify {}, offline broker addr {}", minBrokerId, brokerAddrsNotify, offlineBrokerAddr);
        // 创建请求，请求码：NOTIFY_MIN_BROKER_ID_CHANGE
        RemotingCommand request =
            RemotingCommand.createRequestCommand(RequestCode.NOTIFY_MIN_BROKER_ID_CHANGE, requestHeader);
        // 循环向broker发送最小brokerId发生变化的单向请求
        for (String brokerAddr : brokerAddrsNotify) {
            this.namesrvController.getRemotingClient().invokeOneway(brokerAddr, request, 300);
        }
    }

    /**
     * 如果仅有一个活跃broker，没有指定离线的broker，则直接封装该broker地址返回；
     * 否则，从broker地址表中找出brokerId最小的brokerId；
     * 遍历broker地址表，将不是最小brokerId的broker地址放到集合中，并返回
     * @param brokerAddrMap
     * @param offlineBrokerAddr
     * @return
     */
    private List<String> chooseBrokerAddrsToNotify(Map<Long, String> brokerAddrMap, String offlineBrokerAddr) {
        // 如果仅有一个活跃broker，没有指定离线的broker，则直接封装该broker地址返回
        if (offlineBrokerAddr != null || brokerAddrMap.size() == 1) {
            // notify the reset brokers.
            return new ArrayList<>(brokerAddrMap.values());
        }

        // 从broker地址表中找出brokerId最小的brokerId
        // new broker registered, notify previous brokers.
        long minBrokerId = Collections.min(brokerAddrMap.keySet());
        List<String> brokerAddrList = new ArrayList<>();
        // 遍历broker地址表
        for (Long brokerId : brokerAddrMap.keySet()) {
            // 将不是最小brokerId的broker地址放到集合中
            if (brokerId != minBrokerId) {
                brokerAddrList.add(brokerAddrMap.get(brokerId));
            }
        }
        // 返回
        return brokerAddrList;
    }

    // For test only
    public void printAllPeriodically() {
        try {
            try {
                this.lock.readLock().lockInterruptibly();
                log.info("--------------------------------------------------------");
                {
                    log.info("topicQueueTable SIZE: {}", this.topicQueueTable.size());
                    for (Entry<String, Map<String, QueueData>> next : this.topicQueueTable.entrySet()) {
                        log.info("topicQueueTable Topic: {} {}", next.getKey(), next.getValue());
                    }
                }

                {
                    log.info("brokerAddrTable SIZE: {}", this.brokerAddrTable.size());
                    for (Entry<String, BrokerData> next : this.brokerAddrTable.entrySet()) {
                        log.info("brokerAddrTable brokerName: {} {}", next.getKey(), next.getValue());
                    }
                }

                {
                    log.info("brokerLiveTable SIZE: {}", this.brokerLiveTable.size());
                    for (Entry<BrokerAddrInfo, BrokerLiveInfo> next : this.brokerLiveTable.entrySet()) {
                        log.info("brokerLiveTable brokerAddr: {} {}", next.getKey(), next.getValue());
                    }
                }

                {
                    log.info("clusterAddrTable SIZE: {}", this.clusterAddrTable.size());
                    for (Entry<String, Set<String>> next : this.clusterAddrTable.entrySet()) {
                        log.info("clusterAddrTable clusterName: {} {}", next.getKey(), next.getValue());
                    }
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("printAllPeriodically Exception", e);
        }
    }

    public TopicList getSystemTopicList() {
        TopicList topicList = new TopicList();
        try {
            this.lock.readLock().lockInterruptibly();
            for (Map.Entry<String, Set<String>> entry : clusterAddrTable.entrySet()) {
                topicList.getTopicList().add(entry.getKey());
                topicList.getTopicList().addAll(entry.getValue());
            }

            if (!brokerAddrTable.isEmpty()) {
                for (String s : brokerAddrTable.keySet()) {
                    BrokerData bd = brokerAddrTable.get(s);
                    HashMap<Long, String> brokerAddrs = bd.getBrokerAddrs();
                    if (brokerAddrs != null && !brokerAddrs.isEmpty()) {
                        Iterator<Long> it2 = brokerAddrs.keySet().iterator();
                        topicList.setBrokerAddr(brokerAddrs.get(it2.next()));
                        break;
                    }
                }
            }
        } catch (Exception e) {
            log.error("getSystemTopicList Exception", e);
        } finally {
            this.lock.readLock().unlock();
        }

        return topicList;
    }

    public TopicList getTopicsByCluster(String cluster) {
        TopicList topicList = new TopicList();
        try {
            try {
                this.lock.readLock().lockInterruptibly();
                Set<String> brokerNameSet = this.clusterAddrTable.get(cluster);
                for (String brokerName : brokerNameSet) {
                    for (Entry<String, Map<String, QueueData>> topicEntry : this.topicQueueTable.entrySet()) {
                        String topic = topicEntry.getKey();
                        Map<String, QueueData> queueDataMap = topicEntry.getValue();
                        final QueueData qd = queueDataMap.get(brokerName);
                        if (qd != null) {
                            topicList.getTopicList().add(topic);
                        }
                    }
                }
            } finally {
                this.lock.readLock().unlock();
            }
        } catch (Exception e) {
            log.error("getTopicsByCluster Exception", e);
        }

        return topicList;
    }

    public TopicList getUnitTopics() {
        TopicList topicList = new TopicList();
        try {
            this.lock.readLock().lockInterruptibly();
            for (Entry<String, Map<String, QueueData>> topicEntry : this.topicQueueTable.entrySet()) {
                String topic = topicEntry.getKey();
                Map<String, QueueData> queueDatas = topicEntry.getValue();
                if (queueDatas != null && queueDatas.size() > 0
                    && TopicSysFlag.hasUnitFlag(queueDatas.values().iterator().next().getTopicSysFlag())) {
                    topicList.getTopicList().add(topic);
                }
            }
        } catch (Exception e) {
            log.error("getUnitTopics Exception", e);
        } finally {
            this.lock.readLock().unlock();
        }

        return topicList;
    }

    public TopicList getHasUnitSubTopicList() {
        TopicList topicList = new TopicList();
        try {
            this.lock.readLock().lockInterruptibly();
            for (Entry<String, Map<String, QueueData>> topicEntry : this.topicQueueTable.entrySet()) {
                String topic = topicEntry.getKey();
                Map<String, QueueData> queueDatas = topicEntry.getValue();
                if (queueDatas != null && queueDatas.size() > 0
                    && TopicSysFlag.hasUnitSubFlag(queueDatas.values().iterator().next().getTopicSysFlag())) {
                    topicList.getTopicList().add(topic);
                }
            }
        } catch (Exception e) {
            log.error("getHasUnitSubTopicList Exception", e);
        } finally {
            this.lock.readLock().unlock();
        }

        return topicList;
    }

    public TopicList getHasUnitSubUnUnitTopicList() {
        TopicList topicList = new TopicList();
        try {
            this.lock.readLock().lockInterruptibly();
            for (Entry<String, Map<String, QueueData>> topicEntry : this.topicQueueTable.entrySet()) {
                String topic = topicEntry.getKey();
                Map<String, QueueData> queueDatas = topicEntry.getValue();
                if (queueDatas != null && queueDatas.size() > 0
                    && !TopicSysFlag.hasUnitFlag(queueDatas.values().iterator().next().getTopicSysFlag())
                    && TopicSysFlag.hasUnitSubFlag(queueDatas.values().iterator().next().getTopicSysFlag())) {
                    topicList.getTopicList().add(topic);
                }
            }
        } catch (Exception e) {
            log.error("getHasUnitSubUnUnitTopicList Exception", e);
        } finally {
            this.lock.readLock().unlock();
        }

        return topicList;
    }
}

/**
 * broker地址信息，包括集群名称，broker地址字符串。
 */
class BrokerAddrInfo {
    private String clusterName;
    private String brokerAddr;

    private int hash;

    public BrokerAddrInfo(String clusterName, String brokerAddr) {
        this.clusterName = clusterName;
        this.brokerAddr = brokerAddr;
    }

    public String getClusterName() {
        return clusterName;
    }

    public String getBrokerAddr() {
        return brokerAddr;
    }

    public boolean isEmpty() {
        return clusterName.isEmpty() && brokerAddr.isEmpty();
    }

    @Override
    public boolean equals(Object obj) {
        if (this == obj) {
            return true;
        }
        if (obj == null) {
            return false;
        }

        if (obj instanceof BrokerAddrInfo) {
            BrokerAddrInfo addr = (BrokerAddrInfo) obj;
            return clusterName.equals(addr.clusterName) && brokerAddr.equals(addr.brokerAddr);
        }
        return false;
    }

    @Override
    public int hashCode() {
        int h = hash;
        if (h == 0 && clusterName.length() + brokerAddr.length() > 0) {
            for (int i = 0; i < clusterName.length(); i++) {
                h = 31 * h + clusterName.charAt(i);
            }
            h = 31 * h + '_';
            for (int i = 0; i < brokerAddr.length(); i++) {
                h = 31 * h + brokerAddr.charAt(i);
            }
            hash = h;
        }
        return h;
    }

    @Override
    public String toString() {
        return "BrokerIdentityInfo [clusterName=" + clusterName + ", brokerAddr=" + brokerAddr + "]";
    }
}

/**
 * broker存活信息，包括上次更新时间戳，心跳超时时间（毫秒），数据版本，netty通道实例，高可用服务节点地址
 */
class BrokerLiveInfo {

    /**
     * 上次更新时间戳
     */
    private long lastUpdateTimestamp;

    /**
     * 心跳超时时间，单位毫秒
     */
    private long heartbeatTimeoutMillis;

    /**
     * 数据版本
     */
    private DataVersion dataVersion;

    /**
     * netty的通道实例
     */
    private Channel channel;

    /**
     * 高可用服务节点地址
     */
    private String haServerAddr;

    public BrokerLiveInfo(long lastUpdateTimestamp, long heartbeatTimeoutMillis, DataVersion dataVersion,
        Channel channel,
        String haServerAddr) {
        this.lastUpdateTimestamp = lastUpdateTimestamp;
        this.heartbeatTimeoutMillis = heartbeatTimeoutMillis;
        this.dataVersion = dataVersion;
        this.channel = channel;
        this.haServerAddr = haServerAddr;
    }

    public long getLastUpdateTimestamp() {
        return lastUpdateTimestamp;
    }

    public void setLastUpdateTimestamp(long lastUpdateTimestamp) {
        this.lastUpdateTimestamp = lastUpdateTimestamp;
    }

    public long getHeartbeatTimeoutMillis() {
        return heartbeatTimeoutMillis;
    }

    public void setHeartbeatTimeoutMillis(long heartbeatTimeoutMillis) {
        this.heartbeatTimeoutMillis = heartbeatTimeoutMillis;
    }

    public DataVersion getDataVersion() {
        return dataVersion;
    }

    public void setDataVersion(DataVersion dataVersion) {
        this.dataVersion = dataVersion;
    }

    public Channel getChannel() {
        return channel;
    }

    public void setChannel(Channel channel) {
        this.channel = channel;
    }

    public String getHaServerAddr() {
        return haServerAddr;
    }

    public void setHaServerAddr(String haServerAddr) {
        this.haServerAddr = haServerAddr;
    }

    @Override
    public String toString() {
        return "BrokerLiveInfo [lastUpdateTimestamp=" + lastUpdateTimestamp + ", dataVersion=" + dataVersion
            + ", channel=" + channel + ", haServerAddr=" + haServerAddr + "]";
    }
}

class BrokerStatusChangeInfo {
    Map<Long, String> brokerAddrs;
    String offlineBrokerAddr;
    String haBrokerAddr;

    public BrokerStatusChangeInfo(Map<Long, String> brokerAddrs, String offlineBrokerAddr, String haBrokerAddr) {
        this.brokerAddrs = brokerAddrs;
        this.offlineBrokerAddr = offlineBrokerAddr;
        this.haBrokerAddr = haBrokerAddr;
    }

    public Map<Long, String> getBrokerAddrs() {
        return brokerAddrs;
    }

    public void setBrokerAddrs(Map<Long, String> brokerAddrs) {
        this.brokerAddrs = brokerAddrs;
    }

    public String getOfflineBrokerAddr() {
        return offlineBrokerAddr;
    }

    public void setOfflineBrokerAddr(String offlineBrokerAddr) {
        this.offlineBrokerAddr = offlineBrokerAddr;
    }

    public String getHaBrokerAddr() {
        return haBrokerAddr;
    }

    public void setHaBrokerAddr(String haBrokerAddr) {
        this.haBrokerAddr = haBrokerAddr;
    }
}
